package com.demo.hdfs.fs;

import com.demo.hdfs.FileUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.ArrayList;
import java.util.List;

public class DistributeFileSystemUtil implements FileUtils {
    private static Configuration conf;
    private static  FileSystem fs;

    static {

        System.setProperty("HADOOP_USER_NAME", "hdfs");
        System.setProperty("HADOOP_USER_PASSWORD", "hdfs");
        conf = new Configuration();
        conf.set("fs.defaultFS","hdfs://192.168.150.130:8020");
        conf.set("fs.hdfs.impl","org.apache.hadoop.hdfs.DistributedFileSystem");

        try {
            fs = FileSystem.get(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }

    }

    @Override
    public boolean putHDFS(InputStream file, String hdfsPath) {
        try {
            FSDataOutputStream fsDataOutputStream = fs.create(new Path(hdfsPath));
            IOUtils.copyBytes(file,fsDataOutputStream,1024);
            fsDataOutputStream.hflush();
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
    @Override
    public String getHdfs(String hdfspath){
        FSDataInputStream open = null;
        try {
            open = fs.open(new Path(hdfspath));
            open.seek(0);
            int available = open.available();
            byte[] bytes = new byte[available];
            open.read(bytes);
            return new String(bytes,"utf-8");

        } catch (IOException e) {
            e.printStackTrace();
            return null;
        }
    }
    @Override
    public boolean mkdir(String hdfspath){
        try {
            fs.mkdirs(new Path(hdfspath));
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
        return true;
    }
    @Override
    public List<String> listFile(String hdfspath){
        try {
            List<String> filepaths = new ArrayList<>();
            FileStatus[] fileStatuses = fs.listStatus(new Path(hdfspath));
            for (int i = 0; i < fileStatuses.length; i++) {
                FileStatus fileStatus = fileStatuses[i];
                if(fileStatus.isDirectory()){
                    filepaths.add(fileStatus.getPath().toUri().getPath()+":dir");
                }else {
                    filepaths.add(fileStatus.getPath().toUri().getPath());
                }
            }
            return filepaths;
        } catch (IOException e) {
            e.printStackTrace();
        }
        return null;
    }

    @Override
    public boolean isExist(String hdfs){
        try {
            return fs.exists(new Path(hdfs));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public boolean deletePath(String hdfspath){
        try {
            return fs.deleteOnExit(new Path(hdfspath));
        } catch (IOException e) {
            e.printStackTrace();
        }
        return false;
    }

    @Override
    public void release(){
        try {
            fs.close();
        } catch (IOException e) {
            e.printStackTrace();
        }
    }
}
